package edu.xhu.search.listener;

import edu.xhu.model.search.constant.SearchConstant;
import edu.xhu.utils.common.SearchPrefixUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.data.redis.core.script.RedisScript;
import org.springframework.stereotype.Component;
import org.springframework.data.redis.core.RedisCallback;

import java.util.Arrays;
import java.util.List;
import java.util.Set;

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;

import javax.annotation.PostConstruct;

/**
 * 处理搜索关键字的消费者
 */
@Component
@RequiredArgsConstructor
@Slf4j
public class SearchTermConsumer {

    private final StringRedisTemplate redisTemplate;

    /**
     * 处理搜索条件的 listener，刷新 redis 缓存中的热搜词条
     *
     * @param pushSearchTerms 搜索条件
     */
    @KafkaListener(topics = SearchConstant.HOT_SEARCH_PROCESSING_TOPIC, groupId = SearchConstant.HOT_SEARCH_TOPICS_CONSUMER_GROUP)
    public void handleSearchTerm(List<String> pushSearchTerms,
                                 @Header(KafkaHeaders.RECEIVED_PARTITION_ID) List<Integer> partitions,
                                 @Header(KafkaHeaders.RECEIVED_TOPIC) List<String> topics) {
        for (String pushSearchTerm : pushSearchTerms) {
            if (pushSearchTerm == null || pushSearchTerm.length() < 1)
                return;

            // 1. 刷新热搜词条滑动窗口，list结构，执行 lua 脚本，并拿到被滑动窗口弹出的词条
            String popSearchTerm = redisTemplate.execute(buildHotSearchRefreshLuaScript(),
                    Arrays.asList(SearchConstant.HOT_SEARCH_TERM_WINDOW),
                    SearchConstant.HOT_SEARCH_TERM_WINDOW_SIZE.toString(),
                    pushSearchTerm);


            // 2. 刷新热搜词条排名列表，zset结构，走redis管道，刷新词条的更新时间
            redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                // 2.1 刷新热度排行，zset
                connection.zIncrBy(redisTemplate.getStringSerializer().serialize(SearchConstant.HOT_SEARCH_RANKING_ZSET),
                        1,
                        redisTemplate.getStringSerializer().serialize(pushSearchTerm));

                if (popSearchTerm != null && popSearchTerm.length() > 0)
                    connection.zIncrBy(redisTemplate.getStringSerializer().serialize(SearchConstant.HOT_SEARCH_RANKING_ZSET),
                            -1,
                            redisTemplate.getStringSerializer().serialize(popSearchTerm));


                // 2.2 刷新词条的更新时间，hset
                long timestamp = System.currentTimeMillis();
                connection.hSet(
                        redisTemplate.getStringSerializer().serialize(SearchConstant.HOT_SEARCH_UPDATE_TIME_HSET),
                        redisTemplate.getStringSerializer().serialize(pushSearchTerm),
                        redisTemplate.getStringSerializer().serialize(String.valueOf(timestamp))
                );

                return null;
            });

            // 3. 刷新联想词之前缀树
            SearchPrefixUtil.insertWord(pushSearchTerm);
            SearchPrefixUtil.deleteWord(popSearchTerm);
        }
    }


    /**
     * @return 刷新热搜词条滑动窗口的 lua 脚本
     */
    private RedisScript<String> buildHotSearchRefreshLuaScript() {
        String luaScript = "local list_key = KEYS[1]\n" +
                "local max_size = tonumber(ARGV[1])\n" +
                "local pushElement = ARGV[2]\n" +
                "local popElement\n" +
                "\n" +
                "local list_length = redis.call('LLEN', list_key)\n" +
                "\n" +
                "if list_length < max_size then\n" +
                "    redis.call('RPUSH', list_key, pushElement)\n" +
                "else\n" +
                "    popElement = redis.call('LPOP', list_key)\n" +
                "    redis.call('RPUSH', list_key, pushElement)\n" +
                "end\n" +
                "\n" +
                "if popElement ~= nil then\n" +
                "    return popElement\n" +
                "else\n" +
                "    return nil\n" +
                "end";

        RedisScript<String> script = new DefaultRedisScript<>();
        ((DefaultRedisScript<String>) script).setScriptText(luaScript);
        ((DefaultRedisScript<String>) script).setResultType(String.class);

        return script;
    }

    // lua脚本
    /*
    local list_key = KEYS[1]
    local max_size = tonumber(ARGV[1])
    local pushElement = ARGV[2]
    local popElement

    local list_length = redis.call('LLEN', list_key)

    if list_length < max_size then
        -- 列表未达上限，直接 rpush 元素
        redis.call('RPUSH', list_key, pushElement)
    else
        -- 列表已达上限，先 lpop 一个元素，再 rpush 新元素
        popElement = redis.call('LPOP', list_key)
        redis.call('RPUSH', list_key, pushElement)
    end

    if popElement ~= nil then
         return popElement
    else
         return nil
    end

     */

    @PostConstruct //该方法在 bean 构造后执行，初始化联想词之前缀树，读取 redis 中滑动窗口的全部搜索词
    public void initTrie() {
        Set<String> zsetMembers = redisTemplate.opsForZSet().range(SearchConstant.HOT_SEARCH_RANKING_ZSET, 0, -1);

        if (zsetMembers != null) {
            for (String member : zsetMembers) {
                SearchPrefixUtil.insertWord(member);
            }
        }
    }
}
